package com.whoami.dataplatform.tag_combine.hbase;

import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import com.google.protobuf.Service;
import org.apache.commons.lang.math.NumberUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.protobuf.ResponseConverter;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.List;
import java.util.Stack;

/**
 * @author tzp
 * @since 2019/8/12
 */
@SuppressWarnings("Duplicates")
public class FilterSumEndpoint extends FilterSum.FilterSumService implements Coprocessor, CoprocessorService {
    private static final String TABLE_NAME = "test_tu";
    private static final String CF_DEFAULT = "cf1";

    private RegionCoprocessorEnvironment env;

    @Override
    public Service getService() {
        return this;
    }

    @Override
    public void start(CoprocessorEnvironment env) throws IOException {
        if (env instanceof RegionCoprocessorEnvironment) {
            this.env = (RegionCoprocessorEnvironment) env;
        } else {
            throw new CoprocessorException("Must be loaded on a table region!");
        }
    }

    @Override
    public void stop(CoprocessorEnvironment env) throws IOException {
        // do nothing
    }

    @Override
    public void getSum(RpcController controller, FilterSum.FilterSumRequest request, RpcCallback<FilterSum.FilterSumResponse> done) {
        Scan scan = new Scan();
        scan.addColumn(Bytes.toBytes(CF_DEFAULT), null);
        scan.setMaxVersions(1);

        String[] exps = request.getExp().split(" ");
        FilterSum.FilterSumResponse response = null;
        InternalScanner scanner = null;

        try {
            scanner = env.getRegion().getScanner(scan);
//            scan.setFilter()
            List<Cell> results = new ArrayList<>();
            boolean hasMore = false;
            long sum = 0L;

            do {
                hasMore = scanner.next(results);
                for (Cell cell : results) {
                    BitSet rowkey = BitSet.valueOf(CellUtil.cloneRow(cell));
                    boolean eval = eval(rowkey, exps);
                    if (eval) {
                        sum = sum + Bytes.toInt(CellUtil.cloneValue(cell));
                    }
                }
                results.clear();
            } while (hasMore);

            response = FilterSum.FilterSumResponse.newBuilder().setSum(sum).build();
        } catch (IOException ioe) {
            ResponseConverter.setControllerException(controller, ioe);
        } finally {
            if (scanner != null) {
                try {
                    scanner.close();
                } catch (IOException ignored) {
                }
            }
        }

        done.run(response);
    }

    /**
     * 执行布尔后缀表达式: "24 209 & 24 208 & |"
     */
    public static boolean eval(BitSet bitSet, String[] exps) {
        Stack<Boolean> temp = new Stack<>();
        for (String exp : exps) {
            if (NumberUtils.isNumber(exp)) {
                temp.push(bitSet.get(Integer.parseInt(exp)));
            } else if ("&".equals(exp)) {
                Boolean operand1 = temp.pop();
                Boolean operand2 = temp.pop();
                temp.push(operand1 & operand2);
            } else if ("|".equals(exp)) {
                Boolean operand1 = temp.pop();
                Boolean operand2 = temp.pop();
                temp.push(operand1 | operand2);
            } else if ("^".equals(exp)) {
                Boolean operand1 = temp.pop();
                temp.push(!operand1);
            }
        }
        Boolean result = temp.pop();
        assert temp.size() == 0;
        return result;
    }
}